/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.startup;

import cn.ac.iie.di.ban.ycl.wx.nrd.update.commons.RuntimeEnv;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.commons.RuntimeEnvException;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.exception.UpdateException;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.SenderPool;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.worker.Receiver;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.worker.Sender;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 *
 * @author zwlin
 */
public class WxUpdateStartup {

    public static boolean isTest = false;

    private static SenderPool getSenderPool(String SendNamesrv, String SendTopic) throws REConnectionException {
        RESendSessionBuilder builder = (RESendSessionBuilder) new REConnection(SendNamesrv).getSendSessionBuilder(SendTopic);
        builder.addColumn("wx_url_fng", REFieldType.String, true);
        builder.addColumn("ut", REFieldType.Long, true);
        builder.addColumn("_spec", REFieldType.String, true);
        builder.addColumn("_ch", REFieldType.Long, true);
        builder.addColumn("nrd", REFieldType.Long, true);
        return new SenderPool(builder);
    }

    private static SenderPool getSenderPool2(String SendNamesrv, String SendTopic) throws REConnectionException {
        RESendSessionBuilder builder = (RESendSessionBuilder) new REConnection(SendNamesrv).getSendSessionBuilder(SendTopic);
        builder.addColumn("wx_sfrom", REFieldType.String, true);
        builder.addColumn("nrd", REFieldType.Long, true);
        return new SenderPool(builder);
    }

    private static SenderPool getSenderPool3(String SendNamesrv, String SendTopic) throws REConnectionException {
        RESendSessionBuilder builder = (RESendSessionBuilder) new REConnection(SendNamesrv).getSendSessionBuilder(SendTopic);
        builder.addColumn("_key", REFieldType.String, true);
        builder.addColumn("ut", REFieldType.Long, true);
        builder.addColumn("nrd", REFieldType.Long, true);
        return new SenderPool(builder);
    }

    public static void main(String[] args) throws RuntimeEnvException, UpdateException, REConnectionException {
        RuntimeEnv.init();

//        BlockingQueue<HashMap<String, Object>> updatequeue = new ArrayBlockingQueue<>(500);
        ArrayBlockingQueue<HashMap<String, Object>> queue = new ArrayBlockingQueue<>(10000);
        SenderPool[] pools = new SenderPool[2];
        pools[0] = getSenderPool(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND1_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND1_TOPIC));
        pools[1] = getSenderPool(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND2_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND2_TOPIC));
        SenderPool pooldeltanrd;
        pooldeltanrd = getSenderPool2(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND3_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND3_TOPIC));
        SenderPool toMqPool;
        toMqPool = getSenderPool3(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND4_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_SEND4_TOPIC));

        MongoHelper[] helpers = new MongoHelper[2];
        helpers[0] = new MongoHelper(
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_HOSTS).split(";"),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_DB),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_TABLE),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_USER),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO1_PASSWD));
        helpers[1] = new MongoHelper(
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_HOSTS).split(";"),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_DB),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_TABLE),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_USER),
                RuntimeEnv.getParamAsString(RuntimeEnv.MONGO2_PASSWD));
        Receiver receiver = new Receiver(
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_NAMESRV),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_TOPIC),
                RuntimeEnv.getParamAsString(RuntimeEnv.RMQ_RECEIVE_GROUPNAME_PREFIX),
                pools, pooldeltanrd,
                helpers, queue);
        receiver.start();

        Sender sender = new Sender(pools, pooldeltanrd, helpers, queue, toMqPool);
        sender.start();
    }
}
